package com.gitee.andy.zpush.center;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.gitee.andy.zpush.kernel.SessionContext;
import com.gitee.andy.zpush.kernel.TcpClient;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.util.concurrent.DefaultThreadFactory;

public class ZpushCenterStarter {
	public static void main(String[] args) {
		ConcurrentLinkedQueue<Object> msgQueue = new ConcurrentLinkedQueue<>();

		ScheduledExecutorService ses = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors() * 2,
				new DefaultThreadFactory("write2client-pool"));

		SessionContext sessionContext = SessionContext.create(ses);
		
		sessionContext.add(new TcpClient(1, new ChannelInitializer<SocketChannel>() {
			
			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(
						new ObjectDecoder(1024 * 1024 * 8,
								ClassResolvers.softCachingConcurrentResolver(this.getClass().getClassLoader())),
						new ObjectEncoder(), new CenterClientHandler(sessionContext, msgQueue));
				
			}
		}).connect("127.0.0.1",8008));
		
		ses.scheduleWithFixedDelay(()->{
			Object obj;
			while((obj = msgQueue.poll()) !=null){
				sessionContext.write(obj);
			}
		}, 50, 200, TimeUnit.MILLISECONDS);
	}
}
